package com.gomefinance.producer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.CompletableFuture;

/**
 * @author zhangfuqiang3@gome.com.cn
 * @version 1.0
 * @date 2024/12/28
 * @description:
 **/
@Service
public class KafkaProducerService {
    private static final Logger logger = LoggerFactory.getLogger(KafkaProducerService.class);

    private final KafkaTemplate<String, String> kafkaTemplate;

    public KafkaProducerService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * Send a message to a specific topic.
     *
     * @param topic   the Kafka topic
     * @param key     the key of the message
     * @param message the message payload
     */
    public void sendMessage(String topic, String key, String message) {
        kafkaTemplate.send(topic, key, message);
    }

    /**
     * Send a message to a specific topic without a key.
     *
     * @param topic   the Kafka topic
     * @param message the message payload
     */
    public void  sendMessage(String topic, String message) {
        CompletableFuture<SendResult<String, String>> send = kafkaTemplate.send(topic, message);
        send.whenComplete((result, ex) -> {
            if (ex != null) {
                logger.error("Failed to send message: " + ex.getMessage());
            } else {
                logger.info("已发送");
            }
        });
    }
}
